package operator.condition;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import util.Printer;
import util.RxJavaHelper;
import util.ThreadHelper;

class SkipUntilOperator {

    public static void main(String[] args) {
        skipUntil();
    }

    /***
     * 订阅原始的Observable，但是忽略它的发射物，
     * 直到第二个Observable发射了一项数据那一刻，它开始发射原始Observable
     * 此时第二个Observable立即执行onDispose
     */
    static void skipUntil(){
        Observable.interval(0,1,TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Printer.print("Observable1 doOnNext aLong = " + aLong);
                    }
                })
                .skipUntil(Observable.intervalRange(100,3,2,1, TimeUnit.SECONDS)
                        .doOnNext(new Consumer<Long>() {
                            @Override
                            public void accept(Long aLong) throws Exception {
                                Printer.print("Observable2 doOnNext aLong = " + aLong);
                            }
                        }).doOnDispose(new Action() {
                            @Override
                            public void run() throws Exception {
                                Printer.print("Observable2 doOnDispose");
                            }
                        }))
                .subscribe(RxJavaHelper.getConsumer());
        ThreadHelper.sleep();
    }


}
